package com.atguigu.cn.wc

import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala._

/**
 * @author yangshen
 * @date 2020/4/5 16:59
 */
object StreamWordCount {
  def main(args: Array[String]): Unit = {

    val parameter = ParameterTool.fromArgs(args)
    val host : String = parameter.get("host")
    val port : Int = parameter.getInt("port")

    //创建一个流处理的执行环境
    val environment = StreamExecutionEnvironment.getExecutionEnvironment

    //全局：设置执行环境的并行度
    //environment.setParallelism(1)

    //全局(也可以具体的算子)：禁止任务链的合并，每一个算子 算一个
    //environment.disableOperatorChaining()

    //单个算子：禁止任务链的合并  ->>
    //.filter(_.nonEmpty).disableChaining()

    //设置指定位置合并任务链
    //.filter(_.nonEmpty).startNewChain()

    //接受socket数据流，文本流不可以并发 socketTextStream永远只是一个顺序线行的一个流 单一
    val textDataStream = environment.socketTextStream(host, port)
    //逐一读取数据，打散(分词)之后进行wordCount
    val workCountDataStream = textDataStream.flatMap(_.split("\\s"))
      .filter(_.nonEmpty).startNewChain()
      .map( (_, 1) )
      //分组
      .keyBy(0)
      .sum(1)

    //打印输出
    workCountDataStream.print().setParallelism(1)

    //把输出的内容写到一个文件中
//    workCountDataStream.writeAsText("/te/flink/streamWordCount.text")
    /**
    线程号>：默认最大值是根据当前电脑的cpu数量，也可以设置固定：workCountDataStream.print().setParallelism(1)
    3> (aera,1)
    4> (hea,1)
    2> (awetawetrw,1)
    4> (aer,1)
    4> (atrwae,1)
     */

    //执行任务（作业名字）
    environment.execute("stream word count job")
  }
}
